"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 防篡改
Case Name   : 验证脱敏策略shufflemasking,maskall在资料中体现为弱脱敏函数，实际可以无损的还原数据的问题
Create At   : 2022/3/21
Owner       : @daiguatutu
Description :
    1.设置参数并重启数据库enableseparationofduty=off
    设置参数enable_security_policy = on
    当脱敏策略为maskall时执行以下2-7语句;脱敏策略为shufflemasking时，执行2-10语句;
    2.创建表并插入数据,给某列创建资源标签和脱敏策略
    3.查询脱敏字段
    4.使用类型转换关系查询脱敏字段
    5.使用事务回滚中的delete RETURNING显示原始数据查询脱敏字段
    6.借助with的CTE语法，指定字段查询数据
    7.脱敏字段的子查询和字段拼接查询
    8.gsql使用文件作为命令源,创建函数
    9.调用函数查询数据
    10.调用函数&& 使用字段拼接查询数据
    11.清理环境
Expect      :
    1.成功
    2.成功
    3.成功，查询结果中无zhongguo
    4.查询失败
    5.成功，查询结果中无zhongguo
    6.成功，查询结果中无zhongguo
    7.成功，查询结果中无zhongguo
    8.函数创建成功
    9.成功，查询结果中无zhongguo
    10.成功，查询结果中无zhongguo
    11.清理环境成功
History     :
    Modified by @daiguatutu 2023/02/01: 优化用例,修改8-10步骤的执行条件
    Modified by @daiguatutu 2023/02/07: 修改teardown，删除testscript目录
"""

import os
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class Security(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.pri_node = Node('PrimaryDbUser')
        self.pri_root = Node('PrimaryRoot')
        self.constant = Constant()
        self.common = Common()
        self.t_name = 't_security_desensitization_resume_0001'
        self.masking_policy = 'mp_security_desensitization_resume_0001'
        self.resource_label = 'rl_security_desensitization_resume_0001'
        self.f_name = 'f_security_desensitization_resume_0001'
        self.target_path = os.path.join(macro.DB_INSTANCE_PATH,
                                        'testscript')
        self.default_value1 = self.common.show_param('enableSeparationOfDuty')
        self.default_value2 = self.common.show_param('enable_security_policy')

    def test_security(self):
        text = '-----step1：设置参数enableSeparationOfDuty=off,' \
               'enable_security_policy = on; expect:成功-----'
        self.log.info(text)
        if self.default_value1 != 'off':
            mod_msg = self.pri_sh.execute_gsguc('set',
                                                self.constant.GSGUC_SUCCESS_MSG,
                                                'enableSeparationOfDuty=off')
            self.assertTrue(mod_msg, '执行失败:' + text)
            result = self.pri_sh.restart_db_cluster()
            self.assertTrue(result, '执行失败' + text)
        mod_msg = self.pri_sh.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            'enable_security_policy = on')
        self.assertTrue(mod_msg, '执行失败:' + text)

        masking_policys = ['shufflemasking', 'maskall']
        for policy in masking_policys:
            text = '-----step2：创建表并插入数据,给表创建资源标签和脱敏策略 expect:成功-----'
            self.log.info(text)
            create_cmd = f"drop masking policy if exists " \
                f"{self.masking_policy};" \
                f"drop resource label if exists {self.resource_label};" \
                f"drop table if exists {self.t_name};" \
                f"create table {self.t_name}(" \
                f"pid varchar(148), idcard varchar(148), addr text);" \
                f"insert into {self.t_name} values " \
                f"('0','110289122333','zhongguo jiayou,jiayou');" \
                f"create resource label {self.resource_label} " \
                f"add column({self.t_name}.addr);" \
                f"create masking policy {self.masking_policy} {policy} " \
                f"on label({self.resource_label});"
            msg = self.pri_sh.execut_db_sql(create_cmd)
            self.log.info(msg)
            self.assertIn(self.constant.drop_masking_policy_success, msg,
                          '执行失败:' + text)
            self.assertIn(self.constant.drop_resource_label_success, msg,
                          '执行失败:' + text)
            self.assertIn(self.constant.DROP_TABLE_SUCCESS, msg,
                          '执行失败:' + text)
            self.assertIn(self.constant.CREATE_TABLE_SUCCESS, msg,
                          '执行失败:' + text)
            self.assertIn(self.constant.INSERT_SUCCESS_MSG, msg,
                          '执行失败:' + text)
            self.assertIn(self.constant.resource_label_create_success_msg,
                          msg, '执行失败:' + text)
            self.assertIn(self.constant.masking_policy_create_success_msg,
                          msg, '执行失败:' + text)

            text = '-----step3：查询脱敏字段 expect:成功，查询结果中无zhongguo-----'
            self.log.info(text)
            sql_cmd = f"select addr from {self.t_name};"
            msg = self.pri_sh.execut_db_sql(sql_cmd)
            self.log.info(msg)
            self.assertNotIn('zhongguo', msg, '执行失败:' + text)

            text = '-----step4：使用类型转换关系查询脱敏字段 expect:查询失败-----'
            self.log.info(text)
            sql_cmd = f"select convert_from(addr::bytea,'UTF8') " \
                f"from {self.t_name};"
            msg = self.pri_sh.execut_db_sql(sql_cmd)
            self.log.info(msg)
            self.assertIn('ERROR:  Un-support operation for masking column',
                          msg, '执行失败:' + text)

            text = '-----step5：使用事务回滚中的delete returning显示原始数据 ' \
                   'expect:成功，查询结果中无zhongguo-----'
            self.log.info(text)
            sql_cmd = f"start transaction; " \
                f"delete from {self.t_name} where pid = 0 returning *; " \
                f"rollback;"
            msg = self.pri_sh.execut_db_sql(sql_cmd)
            self.log.info(msg)
            self.assertNotIn('zhongguo', msg, '执行失败:' + text)

            text = '-----step6：借助with的CTE语法，指定字段查询数据 ' \
                   'expect:成功，查询结果中无zhongguo-----'
            self.log.info(text)
            sql_cmd = f"with sub as (select * from {self.t_name}) " \
                f"select * from sub;" \
                f"with sub as (select addr from {self.t_name}) " \
                f"select addr from sub;"
            msg = self.pri_sh.execut_db_sql(sql_cmd)
            self.log.info(msg)
            self.assertNotIn('zhongguo', msg, '执行失败:' + text)

            text = '-----step7：脱敏字段的子查询和字段拼接查询 expect:成功，查询结果中无zhongguo-----'
            self.log.info(text)
            sql_cmd = f"select * from(select addr from {self.t_name});" \
                f"select addr|| '' from {self.t_name};"
            msg = self.pri_sh.execut_db_sql(sql_cmd)
            self.log.info(msg)
            self.assertNotIn('zhongguo', msg, '执行失败:' + text)

            if policy == 'shufflemasking':
                text = '-----step8：gsql使用文件作为命令源,创建函数 expect:函数创建成功-----'
                self.log.info(text)
                self.common.scp_file(self.pri_node,
                                     'Desensitization_Resume_Case0001.sql',
                                     self.target_path)
                sql_file = os.path.join(self.target_path,
                                        'Desensitization_Resume_Case0001.sql')
                sql_cmd = f'source {macro.DB_ENV_PATH} ;' \
                    f'gsql -d {self.pri_node.db_name} ' \
                    f'-p {self.pri_node.db_port} -a -f {sql_file};'
                self.log.info(sql_cmd)
                msg = self.pri_node.sh(sql_cmd).result()
                self.log.info(msg)
                self.assertIn(self.constant.CREATE_FUNCTION_SUCCESS_MSG, msg,
                              '执行失败:' + text)

                text = '-----step9：调用函数查询数据 expect:成功，查询结果中无zhongguo-----'
                self.log.info(text)
                sql_cmd = f"select {self.f_name}({self.f_name}(addr)) " \
                    f"from {self.t_name};"
                msg = self.pri_sh.execut_db_sql(sql_cmd)
                self.log.info(msg)
                self.assertNotIn('zhongguo', msg, '执行失败:' + text)

                text = '-----step10：调用函数&& 使用字段拼接查询数据 ' \
                       'expect:成功，查询结果中无zhongguo-----'
                self.log.info(text)
                sql_cmd = f"select addr,{self.f_name}(addr|| '') " \
                    f"from {self.t_name};"
                msg = self.pri_sh.execut_db_sql(sql_cmd)
                self.log.info(msg)
                self.assertNotIn('zhongguo', msg, '执行失败:' + text)

    def tearDown(self):
        self.log.info('-----step11：清理环境-----')
        text1 = '-----删除函数，标签，脱敏策略，表 expect:成功-----'
        self.log.info(text1)
        drop_cmd = f"drop function if exists {self.f_name};" \
            f"drop masking policy if exists {self.masking_policy};" \
            f"drop resource label if exists {self.resource_label};" \
            f"drop table if exists {self.t_name};"
        drop_msg = self.pri_sh.execut_db_sql(drop_cmd)
        self.log.info(drop_msg)

        text2 = '-----修改参数enableSeparationOfDuty,enable_security_policy ' \
                'expect:成功-----'
        self.log.info(text2)
        mod_msg1 = self.pri_sh.execute_gsguc('set',
                                             self.constant.GSGUC_SUCCESS_MSG,
                                             f'enableSeparationOfDuty='
                                             f'{self.default_value1}')
        restart_res = self.pri_sh.restart_db_cluster()
        mod_msg2 = self.pri_sh.execute_gsguc('reload',
                                             self.constant.GSGUC_SUCCESS_MSG,
                                             f'enable_security_policy='
                                             f'{self.default_value2}')

        text3 = '-----删除testscript目录 expect:成功-----'
        self.log.info(text3)
        del_cmd = f'rm -rf {self.target_path}'
        self.common.get_sh_result(self.pri_root, del_cmd)
        check_cmd = f'if [ -d {self.target_path} ]; ' \
            f'then echo "exists"; else echo "not exists"; fi'
        del_msg = self.common.get_sh_result(self.pri_root, check_cmd)

        self.log.info('-----断言tearDown执行成功-----')
        self.assertIn(self.constant.DROP_FUNCTION_SUCCESS_MSG, drop_msg,
                      '执行失败:' + text1)
        self.assertIn(self.constant.drop_masking_policy_success, drop_msg,
                      '执行失败:' + text1)
        self.assertIn(self.constant.drop_resource_label_success, drop_msg,
                      '执行失败:' + text1)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, drop_msg,
                      '执行失败:' + text1)
        self.assertTrue(mod_msg1, '执行失败:' + text2)
        self.assertTrue(restart_res, '执行失败:' + text2)
        self.assertTrue(mod_msg2, '执行失败:' + text2)
        self.assertEqual('not exists', del_msg, '执行失败:' + text3)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
